package com.lq.rocketmq.transaction;

import com.lq.rocketmq.config.MyConstant;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.concurrent.TimeUnit;

/**
 * @program: myrocketmq
 * @description: 发送事务消息
 * @author: LQ
 * @create: 2021-05-07 17:20
 */

public class TransactionProducer {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //1.创建消息生产者producer,并制定生产者组名,事务消息
        TransactionMQProducer producer = new TransactionMQProducer(MyConstant.MSG_GROUP_TRANSACTION);
        //2.指定Nameserver地址
        producer.setNamesrvAddr(MyConstant.NAMESRV_ADDR);
        // 设置事务消息监听器
        producer.setTransactionListener(new TransactionListener() {
            /**
             * 本地事务执行的方法,在本地事务里面设置事务状态
             * 提交,回滚,未知的状态
             * 对于未知状态会进行回查
             * @param msg 消息
             * @param args 需要事务的消息?
             * @return
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object args) {
                String msgTag = msg.getTags();
                if (StringUtils.equals(msgTag, MyConstant.TAG1)) {
                    // 本地事务提交
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else if (StringUtils.equals(msgTag, MyConstant.TAG2)) {
                    // 本地事务回滚
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } else if (StringUtils.equals(msgTag, MyConstant.TAG3)) {
                    // 未知的状态,进行回查
                    return LocalTransactionState.UNKNOW;
                }
                return LocalTransactionState.UNKNOW;
            }

            /**
             * 本地事务未知状态的回查
             * @param msg
             * @return
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("回查消息的Tag:" + msg.getTags());
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        //3.启动producer
        producer.start();
        String[] tags = {MyConstant.TAG1, MyConstant.TAG2, MyConstant.TAG3};
        //4.创建消息对象,指定主体Topic,Tag和消息体
        for (int i = 0; i < 3; i++) {
            /*
             * 需要三个参数
             * 参数一:消息主题Topic
             * 参数二:消息Tag
             * 参数三:消息内容
             */
            Message msg = new Message(MyConstant.TOPIC_SQL_TRANSACTION, tags[i], ("HelloWorld" + (i + 1)).getBytes());
            // 设置消息自定义属性,用于sql过滤
            msg.putUserProperty("i", String.valueOf(i));
            //5.发送事务控制的消息,2个参数,消息和需要事务的消息,如果为null,即为所有消息
            SendResult result = producer.sendMessageInTransaction(msg, null);
            // 发送状态
            SendStatus status = result.getSendStatus();
            // 消息id
            String msgId = result.getMsgId();
            // 消息接受队列id
            int queueId = result.getMessageQueue().getQueueId();
            System.out.println("发送状态:" + result + ",消息ID" + msgId + ",队列ID" + queueId);
            // 线程睡1秒,因为虚拟机质量原因
            TimeUnit.SECONDS.sleep(1);
        }
        //6.关闭生产者producer
        // 因为事务消息需要回查,如果发完消息直接关闭,会导致没办法进行回查
//        producer.shutdown();
    }

}
